package com.baijiaoxi.common.model.asyncjob.eventbus;

import com.baijiaoxi.common.model.asyncjob.AsyncJobProducer;
import com.baijiaoxi.common.model.asyncjob.AsyncJobPublishCallback;
import com.baijiaoxi.common.model.mq.MQBody;
import com.google.common.eventbus.EventBus;

public class EventProducer implements AsyncJobProducer {

    private EventBusManager eventBusManager;
    private String prefix;

    public EventProducer(EventBusManager eventBusManager, String prefix) {
        this.eventBusManager = eventBusManager;
        this.prefix = prefix;
    }


    @Override
    public void publish(String topic, MQBody data, AsyncJobPublishCallback asyncJobPublishCallback) {
        String finalTopic = prefix + topic;
        doPublish(finalTopic, data, asyncJobPublishCallback);
    }

    @Override
    public void republish(String topic, MQBody data, AsyncJobPublishCallback asyncJobPublishCallback) {
        doPublish(topic, data, asyncJobPublishCallback);
    }

    public void doPublish(String topic, MQBody data, AsyncJobPublishCallback asyncJobPublishCallback) {
        try {
            EventBus eventBus = eventBusManager.getEventBus(topic);
            eventBus.post(data);
            EventStorage eventStorage = eventBusManager.getEventStorage();
            if (eventStorage != null)
                eventStorage.in(topic, data);
            asyncJobPublishCallback.onSuccess(data);
        } catch (Exception e) {
            asyncJobPublishCallback.onException(e);
        }
    }
}
